package cn.felord.debezium.config;

import io.debezium.connector.mysql.MySqlConnector;
import io.debezium.data.Envelope;
import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static java.util.stream.Collectors.toMap;


@Configuration
public class DebeziumConfiguration {


    @Value("${debezium.offset.storage.file.filename}")
    private String offsetStorageFilename;

    @Value("${debezium.offset.flush.interval.ms}")
    private String offsetFlushIntervalMs;

    @Value("${debezium.name}")
    private String name;

    @Value("${debezium.database.hostname}")
    private String dbHostname;

    @Value("${debezium.database.port}")
    private String dbPort;

    @Value("${debezium.database.user}")
    private String dbUser;

    @Value("${debezium.database.password}")
    private String dbPassword;

    @Value("${debezium.database.include.list}")
    private String databaseIncludeList;

    @Value("${debezium.table.include.list}")
    private String tableIncludeList;

    @Value("${debezium.include.schema.changes}")
    private String includeSchemaChanges;

    @Value("${debezium.database.server.id}")
    private String serverId;

    @Value("${debezium.database.server.name}")
    private String serverName;

    @Value("${debezium.database.history.file.filename}")
    private String databaseHistoryFilename;

    @Bean
    io.debezium.config.Configuration debeziumConfig() {
        return io.debezium.config.Configuration.create()
//            连接器的Java类名称
                .with("connector.class", MySqlConnector.class.getName())
//            偏移量持久化，用来容错 默认值
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
//                偏移量持久化文件路径 默认/tmp/offsets.dat  如果路径配置不正确可能导致无法存储偏移量 可能会导致重复消费变更
//                如果连接器重新启动，它将使用最后记录的偏移量来知道它应该恢复读取源信息中的哪个位置。
                .with("offset.storage.file.filename", offsetStorageFilename)
//                捕获偏移量的周期
                .with("offset.flush.interval.ms", offsetFlushIntervalMs)
//               连接器的唯一名称
                .with("name", name)
//                数据库的hostname
                .with("database.hostname", dbHostname)
//                端口
                .with("database.port", dbPort)
//                用户名
                .with("database.user", dbUser)
//                密码
                .with("database.password", dbPassword)
//                 包含的数据库列表
                .with("database.include.list", databaseIncludeList)
                .with("table.include.list", tableIncludeList)
//                .with("topic.prefix", " myprefix")
//                是否包含数据库表结构层面的变更，建议使用默认值true
                .with("include.schema.changes", includeSchemaChanges)
//                mysql.cnf 配置的 server-id
                .with("database.server.id", serverId)
//                	MySQL 服务器或集群的逻辑名称
                .with("database.server.name", serverName)
//                历史变更记录
                .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
//                历史变更记录存储位置，存储DDL
                .with("database.history.file.filename", databaseHistoryFilename)
                .build();
    }

    /**
     * Debezium server bootstrap debezium server bootstrap.
     *
     * @param configuration the configuration
     * @return the debezium server bootstrap
     */
    @Bean
    DebeziumServerBootstrap debeziumServerBootstrap(io.debezium.config.Configuration configuration) {
        DebeziumServerBootstrap debeziumServerBootstrap = new DebeziumServerBootstrap();
        DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(configuration.asProperties())
                .notifying(this::handlePayload)
                .build();

        debeziumServerBootstrap.setDebeziumEngine(debeziumEngine);
        return debeziumServerBootstrap;
    }


    private Map<String, Object> payload(Struct struct) {
        if (struct == null){
            return null;
        }
        return struct.schema().fields().stream()
                .map(Field::name)
                .filter(fieldName -> struct.get(fieldName) != null)
                .collect(toMap(fieldName -> fieldName, fieldName -> struct.get(fieldName)));

    }

    private void handlePayload(List<RecordChangeEvent<SourceRecord>> recordChangeEvents, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> recordCommitter) {
        for (RecordChangeEvent<SourceRecord> r : recordChangeEvents) {
            SourceRecord sourceRecord = r.record();
            Struct sourceRecordChangeValue = (Struct) sourceRecord.value();

            if (sourceRecordChangeValue != null) {
                Envelope.Operation operation = Envelope.Operation.forCode((String) sourceRecordChangeValue.get(Envelope.FieldName.OPERATION));
                // 判断操作类型并过滤掉READ操作
                if (operation != Envelope.Operation.READ) {
                    Map<String, Object> resMap = new HashMap<>();
                    resMap.put("operation", operation);
                    // 删除操作
                    Struct beforestruct = (Struct) sourceRecordChangeValue.get(Envelope.FieldName.BEFORE);
                    Struct afterstruct = (Struct) sourceRecordChangeValue.get(Envelope.FieldName.AFTER);
                    switch (operation) {
                        case DELETE:
                            resMap.put("before", payload(beforestruct));
                            resMap.put("after", null);
                            break;
                        case CREATE:
                            resMap.put("before", null);
                            resMap.put("after", payload(afterstruct));
                            break;
                        case UPDATE:
                            resMap.put("before", payload(beforestruct));
                            resMap.put("after", payload(afterstruct));
                            break;
                        default:
                            // 可以选择在这里处理未匹配的情况，或者省略 default 分支
                            break;
                    }

                    System.out.println("Payload: " + resMap);
                }
            }

            try {
                // 提交更改事件
                recordCommitter.markProcessed(r);
            } catch (Exception e) {
                System.err.println("Failed to commit record change event: " + e.getMessage());
            }
        }
    }

}
